package com.wzf.flink.starter;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.wzf.properies.CustomizerFlinkProperties;
import com.wzf.task.sink.CustomerSink;
import com.wzf.utils.SavepointUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
@EnableConfigurationProperties({CustomizerFlinkProperties.class})
@Slf4j
public class FlinkStarter implements ApplicationRunner {

    @Resource
    private CustomizerFlinkProperties properties;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        startFlink();
    }

    private void  startFlink() throws Exception {
        String savepointDir = SavepointUtils.getSavepointRestore("/home/"+properties.getSavePointDir());
        Configuration configuration = new Configuration();
        configuration.setInteger(RestOptions.PORT, properties.getWebuiPort());
        if (savepointDir != null) {
            SavepointRestoreSettings savepointRestoreSettings = SavepointRestoreSettings.forPath(savepointDir);
            SavepointRestoreSettings.toConfiguration(savepointRestoreSettings, configuration);
        }
        MySqlSource mySqlSource = mysqlConfig();
        doStartFlink(configuration,mySqlSource);
    }


    private void  doStartFlink(Configuration configuration,MySqlSource mySqlSource) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        env.setParallelism(1);
        env.enableCheckpointing(1000L);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();

        //1.设置Checkpoint间隔时间,单位毫秒
        env.enableCheckpointing(2000);
        //2.设置state状态后端，内存存储
        env.setStateBackend(new HashMapStateBackend());
        //获取env的CheckpointConfig对象
        //3.检查点存储
        //JM内存存储
        //ckptConfig.setCheckpointStorage(new JobManagerCheckpointStorage)
        //HDFS文件系统存储(暂不演示)
        //ckptConfig.setCheckpointStorage("hdfs://mycluster/flink/ckpts")
        //本地文件系统存储
        //4.设置checkpoint执行模式 默认EXACTLY_ONCE精准一次,需要Source和Sink的支持
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //5.设置两个checkpoint之间最少的等待时间  单位毫秒
        checkpointConfig.setMinPauseBetweenCheckpoints(500);
        //5.设置同一时刻允许多少个checkpoint同时执行
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        //6.设置checkpoint超时时间  如果一次ckpt在30s内还不能完成，说明本次失败则丢弃
        checkpointConfig.setCheckpointTimeout(30000);
        //7.设置允许ckpt失败次数
        checkpointConfig.setTolerableCheckpointFailureNumber(3);
        //8.设置当Cancel当前job时是否清理检查点，默认 Checkpoint会在作业被Cancel时被清理(被删除）
        //DELETE_ON_CANCELLATION,当作业被取消时，删除外部的checkpoint(默认值)
        //RETAIN_ON_CANCELLATION，当作业被取消时，保留外部的checkpoint
        //checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        checkpointConfig.setCheckpointStorage("file:///home/"+properties.getCheckoutDir());
        //9.设置重启策略  noRestart
        env.setRestartStrategy(RestartStrategies.fallBackRestart());
        DataStreamSource<String> cdcSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), properties.getDialect());
        cdcSource.addSink(new CustomerSink()).setParallelism(1).name("mysql-");
        env.execute(FlinkStarter.class.getName());

    }

    private MySqlSource mysqlConfig(){

        return MySqlSource.<String>builder()
                .hostname(properties.getHost())
                .port(properties.getPort())
                .databaseList(properties.getDbs().split(",")) // set captured database
                .tableList(properties.getTables().split(",")) // set captured table
                .username(properties.getUsername())
                .password(properties.getPassword())
                .serverId(properties.getServerId())
                //.serverTimeZone("connectionTimeZone")
                .startupOptions(StartupOptions.latest()) // 启动连接器后, 才开始采集第一条数据
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();
    }

}
